package com.example.yckjbigdataservice.utils;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * HBase异步操作完整示例
 * 包含连接管理、CRUD操作和资源安全关闭
 */
public class HBaseAsyncExample {

    private static final Logger logger = LoggerFactory.getLogger(HBaseAsyncExample.class);

    // 连接配置常量
    private static final String ZOOKEEPER_QUORUM = "localhost:2181";
    private static final String TABLE_NAME = "test_table";
    private static final String COLUMN_FAMILY = "cf1";
    private static final String COLUMN_QUALIFIER = "col1";

    // 超时配置
    private static final long OPERATION_TIMEOUT = 30000; // 30秒
    private static final long CONNECTION_TIMEOUT = 60000; // 60秒

    private AsyncConnection asyncConnection;
    private AsyncTable<AdvancedScanResultConsumer> asyncTable;

    /**
     * 初始化HBase连接
     */
    public void initConnection() throws IOException, ExecutionException, InterruptedException, TimeoutException {
        logger.info("正在初始化HBase异步连接...");

        // 配置HBase连接参数
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", ZOOKEEPER_QUORUM);
        conf.set("hbase.zookeeper.property.clientPort", "2181");

        // 连接池配置，提升性能
        conf.setInt("hbase.client.max.total.tasks", 200);
        conf.setInt("hbase.client.max.perserver.tasks", 20);
        conf.setInt("hbase.client.max.perregion.tasks", 10);

        // 重试和超时配置
        conf.setInt("hbase.client.retries.number", 3);
        conf.setLong("hbase.client.operation.timeout", OPERATION_TIMEOUT);
        conf.setLong("hbase.client.scanner.timeout.period", OPERATION_TIMEOUT);

        // 创建异步连接
        this.asyncConnection = ConnectionFactory.createAsyncConnection(conf)
                .get(CONNECTION_TIMEOUT, TimeUnit.MILLISECONDS);

        // 获取异步表
        this.asyncTable = asyncConnection.getTable(TableName.valueOf(TABLE_NAME));

        logger.info("HBase异步连接初始化成功");
    }

    /**
     * 异步插入数据
     */
    public CompletableFuture<Void> putDataAsync(String rowKey, String value) {
        logger.info("异步插入数据: rowKey={}, value={}", rowKey, value);

        Put put = new Put(Bytes.toBytes(rowKey));
        put.addColumn(Bytes.toBytes(COLUMN_FAMILY),
                Bytes.toBytes(COLUMN_QUALIFIER),
                Bytes.toBytes(value));

        return asyncTable.put(put)
                .whenComplete((result, throwable) -> {
                    if (throwable != null) {
                        logger.error("异步插入数据失败: rowKey={}", rowKey, throwable);
                    } else {
                        logger.info("异步插入数据成功: rowKey={}", rowKey);
                    }
                });
    }

    /**
     * 异步获取数据
     */
    public CompletableFuture<String> getDataAsync(String rowKey) {
        logger.info("异步获取数据: rowKey={}", rowKey);

        Get get = new Get(Bytes.toBytes(rowKey));
        get.addColumn(Bytes.toBytes(COLUMN_FAMILY), Bytes.toBytes(COLUMN_QUALIFIER));

        return asyncTable.get(get)
                .thenApply(result -> {
                    if (result.isEmpty()) {
                        logger.warn("数据不存在: rowKey={}", rowKey);
                        return null;
                    }

                    byte[] value = result.getValue(Bytes.toBytes(COLUMN_FAMILY),
                            Bytes.toBytes(COLUMN_QUALIFIER));
                    String valueStr = value != null ? Bytes.toString(value) : null;
                    logger.info("异步获取数据成功: rowKey={}, value={}", rowKey, valueStr);
                    return valueStr;
                })
                .exceptionally(throwable -> {
                    logger.error("异步获取数据失败: rowKey={}", rowKey, throwable);
                    return null;
                });
    }

    /**
     * 异步删除数据
     */
    public CompletableFuture<Void> deleteDataAsync(String rowKey) {
        logger.info("异步删除数据: rowKey={}", rowKey);

        Delete delete = new Delete(Bytes.toBytes(rowKey));

        return asyncTable.delete(delete)
                .whenComplete((result, throwable) -> {
                    if (throwable != null) {
                        logger.error("异步删除数据失败: rowKey={}", rowKey, throwable);
                    } else {
                        logger.info("异步删除数据成功: rowKey={}", rowKey);
                    }
                });
    }

    /**
     * 批量异步操作示例
     */
    public CompletableFuture<Void> batchOperationsAsync() {
        logger.info("执行批量异步操作");

        // 批量插入
        CompletableFuture<Void> put1 = putDataAsync("row1", "value1");
        CompletableFuture<Void> put2 = putDataAsync("row2", "value2");
        CompletableFuture<Void> put3 = putDataAsync("row3", "value3");

        // 等待批量插入完成后，再进行批量查询
        return CompletableFuture.allOf(put1, put2, put3)
                .thenCompose(v -> {
                    logger.info("批量插入完成，开始批量查询");

                    CompletableFuture<String> get1 = getDataAsync("row1");
                    CompletableFuture<String> get2 = getDataAsync("row2");
                    CompletableFuture<String> get3 = getDataAsync("row3");

                    return CompletableFuture.allOf(get1, get2, get3)
                            .thenRun(() -> logger.info("批量操作全部完成"));
                });
    }

    /**
     * 检查连接是否正常
     */
    public boolean isConnectionValid() {
        try {
            return asyncConnection != null && !asyncConnection.isClosed();
        } catch (Exception e) {
            logger.error("检查连接状态失败", e);
            return false;
        }
    }

    /**
     * 安全关闭连接
     * 注意：只需要关闭AsyncConnection，AsyncTable会自动失效
     */
    public void closeConnection() {
        logger.info("正在关闭HBase连接...");

        try {
            if (asyncConnection != null && !asyncConnection.isClosed()) {
                // 给异步操作一些时间完成
                Thread.sleep(1000);

                // 只需要关闭AsyncConnection
                asyncConnection.close();

                // AsyncTable会自动失效，不需要手动关闭
                asyncTable = null;

                logger.info("HBase连接已安全关闭");
            }
        } catch (Exception e) {
            logger.error("关闭HBase连接时发生错误", e);
        }
    }

    /**
     * 主方法示例
     */
    public static void main(String[] args) {
        HBaseAsyncExample example = new HBaseAsyncExample();

        try {
            // 初始化连接
            example.initConnection();

            // 验证连接
            if (!example.isConnectionValid()) {
                logger.error("HBase连接无效");
                return;
            }

            // 执行单个操作示例
            example.putDataAsync("test_row", "test_value")
                    .thenCompose(v -> example.getDataAsync("test_row"))
                    .thenAccept(value -> logger.info("获取到的值: {}", value))
                    .get(OPERATION_TIMEOUT, TimeUnit.MILLISECONDS);

            // 执行批量操作示例
            example.batchOperationsAsync()
                    .get(OPERATION_TIMEOUT * 3, TimeUnit.MILLISECONDS);

            // 清理测试数据
            example.deleteDataAsync("test_row")
                    .thenCompose(v -> example.deleteDataAsync("row1"))
                    .thenCompose(v -> example.deleteDataAsync("row2"))
                    .thenCompose(v -> example.deleteDataAsync("row3"))
                    .get(OPERATION_TIMEOUT, TimeUnit.MILLISECONDS);

            logger.info("所有操作执行完成");

        } catch (Exception e) {
            logger.error("执行HBase操作时发生错误", e);
        } finally {
            // 确保资源被正确释放
            example.closeConnection();
        }
    }
}

/**
 * 连接管理工具类 - 单例模式管理连接
 */
class HBaseConnectionManager {
    private static final Logger logger = LoggerFactory.getLogger(HBaseConnectionManager.class);
    private static volatile HBaseConnectionManager instance;
    private volatile AsyncConnection asyncConnection;
    private final Object lock = new Object();

    private HBaseConnectionManager() {}

    public static HBaseConnectionManager getInstance() {
        if (instance == null) {
            synchronized (HBaseConnectionManager.class) {
                if (instance == null) {
                    instance = new HBaseConnectionManager();
                }
            }
        }
        return instance;
    }

    public AsyncConnection getConnection() throws IOException, ExecutionException, InterruptedException, TimeoutException {
        if (asyncConnection == null || asyncConnection.isClosed()) {
            synchronized (lock) {
                if (asyncConnection == null || asyncConnection.isClosed()) {
                    createConnection();
                }
            }
        }
        return asyncConnection;
    }

    private void createConnection() throws IOException, ExecutionException, InterruptedException, TimeoutException {
        logger.info("创建新的HBase连接");

        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "localhost:2181");

        // 性能优化配置
        conf.setInt("hbase.client.max.total.tasks", 200);
        conf.setInt("hbase.client.max.perserver.tasks", 20);
        conf.setLong("hbase.client.operation.timeout", 30000);

        this.asyncConnection = ConnectionFactory.createAsyncConnection(conf)
                .get(60000, TimeUnit.MILLISECONDS);
    }

    public void closeConnection() {
        if (asyncConnection != null) {
            try {
                asyncConnection.close();
                logger.info("HBase连接已关闭");
            } catch (IOException e) {
                logger.error("关闭HBase连接失败", e);
            }
        }
    }
}
